﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace BookSleeve
{
    /// <summary>
    /// Base class for a redis-connection; provides core redis services
    /// </summary>
    public abstract class RedisConnectionBase : IDisposable
    {
        AsyncMessageSocket asyncSocket;

        private readonly MessageQueue unsent;
        private readonly int port, ioTimeout, syncTimeout;
        private readonly string host, password;
        /// <summary>
        /// The amount of time to wait for any individual command to return a result when using Wait
        /// </summary>
        public int SyncTimeout { get { return syncTimeout; } }
        /// <summary>
        /// The host for the redis server
        /// </summary>
        public string Host { get { return host; } }
        /// <summary>
        /// The password used to authenticate with the redis server
        /// </summary>
        protected string Password { get { return password; } }
        /// <summary>
        /// The port for the redis server
        /// </summary>
        public int Port { get { return port; } }
        /// <summary>
        /// The IO timeout to use when communicating with the redis server
        /// </summary>
        protected int IOTimeout { get { return ioTimeout; } }
        private RedisFeatures features;
        /// <summary>
        /// Features available to the redis server
        /// </summary>
        public virtual RedisFeatures Features { get { return features; } }
        /// <summary>
        /// The version of the connected redis server
        /// </summary>
        public virtual Version ServerVersion
        {
            get
            {
                var tmp = features;
                return tmp == null ? null : tmp.Version;
            }
            private set
            {
                features = new RedisFeatures(value);
            }
        }
        /// <summary>
        /// Obtains fresh statistics on the usage of the connection
        /// </summary>
        protected void GetCounterValues(out int messagesSent, out int messagesReceived,
            out int queueJumpers, out int messagesCancelled, out int unsent, out int errorMessages, out int timeouts)
        {
            messagesSent = Interlocked.CompareExchange(ref this.messagesSent, 0, 0);
            messagesReceived = Interlocked.CompareExchange(ref this.messagesReceived, 0, 0);
            queueJumpers = Interlocked.CompareExchange(ref this.queueJumpers, 0, 0);
            messagesCancelled = Interlocked.CompareExchange(ref this.messagesCancelled, 0, 0);
            messagesSent = Interlocked.CompareExchange(ref this.messagesSent, 0, 0);
            errorMessages = Interlocked.CompareExchange(ref this.errorMessages, 0, 0);
            timeouts = Interlocked.CompareExchange(ref this.timeouts, 0, 0);
            unsent = this.unsent.GetCount();
        }
        /// <summary>
        /// Issues a basic ping/pong pair against the server, returning the latency
        /// </summary>
        protected Task<long> Ping(bool queueJump)
        {
            return ExecuteInt64(new PingMessage(), queueJump);
        }
        /// <summary>
        /// The default time to wait for individual commands to complete when using Wait
        /// </summary>
        protected const int DefaultSyncTimeout = 10000;
        // dont' really want external subclasses
        internal RedisConnectionBase(string host, int port = 6379, int ioTimeout = -1, string password = null,
            int syncTimeout = DefaultSyncTimeout)
        {
            if(syncTimeout <= 0) throw new ArgumentOutOfRangeException("syncTimeout");
            this.syncTimeout = syncTimeout;
            this.unsent = new MessageQueue();
            this.host = host;
            this.port = port;
            this.ioTimeout = ioTimeout;
            this.password = password;

            IncludeDetailInTimeouts = true;
        }
        static bool TryParseVersion(string value, out Version version)
        {  // .NET 4.0 has Version.TryParse, but 3.5 CP does not
            try
            {
                version = new Version(value);
                return true;
            }
            catch
            {
                version = default(Version);
                return false;
            }
        }

        private int state;
        /// <summary>
        /// The current state of the connection
        /// </summary>
        public virtual ConnectionState State
        {
            get { return (ConnectionState)state; }
        }
        /// <summary>
        /// Releases any resources associated with the connection
        /// </summary>
        public virtual void Dispose()
        {
            try { Close(!FlushOnDispose); }
            catch { }
            
            abort = true;
            ResetKeepAlive(); // abandon any timers, if any
            Error = null;         
        }
        /// <summary>
        /// Called after opening a connection
        /// </summary>
        protected virtual void OnOpened() { }
        /// <summary>
        /// Called before opening a connection
        /// </summary>
        protected virtual void OnOpening() { }

        /// <summary>
        /// Called during connection init, but after the AUTH is sent (if needed)
        /// </summary>
        protected virtual void OnInitConnection() { }

        /// <summary>
        /// Should Dispose block on unsent commands?
        /// </summary>
        public bool FlushOnDispose { get; set; }

        /// <summary>
        /// Attempts to open the connection to the remote server
        /// </summary>
        public Task Open()
        {
            if (Interlocked.CompareExchange(ref state, (int)ConnectionState.Opening, (int)ConnectionState.Shiny) != (int)ConnectionState.Shiny)
                throw new InvalidOperationException(); // not shiny
            try
            {
                OnOpening();

                asyncSocket = new AsyncMessageSocket(this);
                asyncSocket.ReadTimeout = asyncSocket.WriteTimeout = ioTimeout;
                ThreadPool.QueueUserWorkItem(x =>
                {
                    try
                    {
                        ((AsyncMessageSocket)x).Open(new DnsEndPoint(host, port));
                    }
                    catch (Exception ex)
                    {
                        OnError("Open", ex, true);
                    }
                }, asyncSocket);

                if (!string.IsNullOrEmpty(password)) EnqueueMessage(RedisMessage.Create(-1, RedisLiteral.AUTH, password).ExpectOk().Critical(), true);

                var info = GetInfo();
                OnInitConnection();
                Flush();

                return ContinueWith(info, done =>
                {
                    switch(done.Status)
                    {
                        case TaskStatus.RanToCompletion:
                            try
                            {
                                // process this when available
                                var parsed = ParseInfo(done.Result);
                                string s;
                                Version version;
                                if (parsed.TryGetValue("redis_version", out s) && TryParseVersion(s, out version))
                                {
                                    this.ServerVersion = version;
                                }
                                if (parsed.TryGetValue("redis_mode", out s) && s == "sentinel")
                                {
                                    ServerType = BookSleeve.ServerType.Sentinel;
                                }
                                else if (parsed.TryGetValue("role", out s) && s != null)
                                {
                                    switch (s)
                                    {
                                        case "master": ServerType = BookSleeve.ServerType.Master; break;
                                        case "slave": ServerType = BookSleeve.ServerType.Slave; break;
                                    }
                                }
                                Interlocked.CompareExchange(ref state, (int)ConnectionState.Open, (int)ConnectionState.Opening);
                            }
                            catch
                            {
                                Close(true);
                                Interlocked.CompareExchange(ref state, (int)ConnectionState.Closed, (int)ConnectionState.Opening);
                            }
                            break;
                        case TaskStatus.Faulted:
                            throw done.Exception;
                    }
                });
            }
            catch
            {
                Interlocked.CompareExchange(ref state, (int)ConnectionState.Closed, (int)ConnectionState.Opening);
                throw;
            }
        }

        /// <summary>
        /// Define any command re-mappings defined on the server; mapping a command to a null or empty string deactivates the command
        /// </summary>
        public void SetCommandMap(Dictionary<string, string> map)
        {
            if (map == null || map.Count == 0) commandMap = defaultCommandMap;
            else
            {
                byte[][] newMap = (byte[][])defaultCommandMap.Clone();

                bool anyDelta = false;
                foreach(var pair in map)
                {
                    if(string.IsNullOrEmpty(pair.Key)) continue;
                    RedisLiteral key;
                    if(!Enum.TryParse<RedisLiteral>(pair.Key, true, out key)) continue; // ignore commands we don't know
                    byte[] val;
                    bool localDelta = false;
                    if (string.IsNullOrWhiteSpace(pair.Value))
                    {
                        val = null;
                        localDelta = true;
                    }
                    else
                    {
                        val = Encoding.ASCII.GetBytes(pair.Value.Trim());
                        byte[] defaultVal = defaultCommandMap[(int)key];
                        if(defaultVal.Length != val.Length)
                        {
                            localDelta = true;
                        }
                        else
                        {
                            for(int i = 0 ; i < val.Length ; i++)
                            {
                                if(val[i] != defaultVal[i]) 
                                {
                                    localDelta = true;
                                    break;
                                }
                            }
                        }
                    }
                    if (localDelta)
                    {
                        newMap[(int)key] = val;
                        anyDelta = true;
                    }
                }
                commandMap = anyDelta ? newMap : defaultCommandMap;
            }
        }
        private byte[][] commandMap = defaultCommandMap;

        private static readonly byte[][] defaultCommandMap;
        static RedisConnectionBase()
        {
            var arr = Enum.GetValues(typeof(RedisLiteral));
            defaultCommandMap = new byte[arr.Length][];
            foreach (RedisLiteral literal in arr)
            {
                defaultCommandMap[(int)literal] = Encoding.ASCII.GetBytes(literal.ToString().ToUpperInvariant());
            }
        }

        /// <summary>
        /// The INFO command returns information and statistics about the server in format that is simple to parse by computers and easy to red by humans.
        /// </summary>
        /// <remarks>http://redis.io/commands/info</remarks>
        public Task<string> GetInfo(bool queueJump = false)
        {
            return GetInfo(null, queueJump);
        }
        /// <summary>
        /// The INFO command returns information and statistics about the server in format that is simple to parse by computers and easy to red by humans.
        /// </summary>
        /// <remarks>http://redis.io/commands/info</remarks>
        public Task<string> GetInfo(string category, bool queueJump = false)
        {
            var msg = string.IsNullOrEmpty(category) ? RedisMessage.Create(-1, RedisLiteral.INFO) : RedisMessage.Create(-1, RedisLiteral.INFO, category);
            return ExecuteString(msg, queueJump);
        }
        static Dictionary<string, string> ParseInfo(string result)
        {
            string[] lines = result.Split(new[] { "\r\n" }, StringSplitOptions.RemoveEmptyEntries);
            var data = new Dictionary<string, string>();
            for (int i = 0; i < lines.Length; i++)
            {
                string line = lines[i];
                if (string.IsNullOrEmpty(line) || line[0] == '#') continue; // 2.6+ can have empty lines, and comment lines
                int idx = line.IndexOf(':');
                if (idx > 0) // double check this line looks about right
                {
                    data.Add(line.Substring(0, idx), line.Substring(idx + 1));
                }
            }
            return data;
        }

        int timeouts;

        /// <summary>
        /// Indicate the number of messages that have not yet been set.
        /// </summary>
        public virtual int OutstandingCount { get { return unsent.GetCount(); } }
        
        /// <summary>
        /// Raised when a connection becomes closed.
        /// </summary>
        public event EventHandler Closed;
        volatile bool abort;
        /// <summary>
        /// Closes the connection; either draining the unsent queue (to completion), or abandoning the unsent queue.
        /// </summary>
        public void Close(bool abort)
        {
            LogToConsole("(closing while {0})", (ConnectionState)state);
            this.abort = abort;
            Task quit = null;
            switch ((ConnectionState)state)
            {
                case ConnectionState.Closing:
                case ConnectionState.Closed:
                    break; // nothing to do
                case ConnectionState.Opening:
                case ConnectionState.Open:
                    if(!abort) quit = ExecuteVoid(RedisMessage.Create(-1, RedisLiteral.QUIT), false);
                    Interlocked.Exchange(ref state, (int)ConnectionState.Closing);
                    break;
                case ConnectionState.Shiny:
                default:
                    Interlocked.Exchange(ref state, (int)ConnectionState.Closed); // kill it dead
                    break;
            }
            ResetKeepAlive(); // abandon any timers, if any
            if (quit != null)
            {
                try
                {
                    Wait(quit);
                }
                catch (Exception ex)
                {
                    OnError("Close", ex, true);
                }
            }
            try { if (asyncSocket != null) asyncSocket.Dispose(); }
            catch { }
            asyncSocket = null;
        }
        internal virtual void ResetKeepAlive() { }

        internal virtual object ProcessReply(ref RedisResult result)
        {
            RedisMessage message;
            lock (sent)
            {
                int count = sent.Count;
                if (count == 0) throw new RedisException("Data received with no matching message");
                message = sent.Dequeue();
                if (count == 1) Monitor.Pulse(sent); // in case the outbound stream is closing and needs to know we're up-to-date
            }
            return ProcessReply(ref result, message);
        }
        internal bool IsAborting
        {
            get { return abort; }
        }
        internal virtual object ProcessReply(ref RedisResult result, RedisMessage message)
        {
            byte[] expected;
            if (!result.IsError && (expected = message.GetExpected(commandMap)) != null)
            {
                result = result.IsMatch(expected)
                ? RedisResult.Pass : RedisResult.Error(result.ValueString);
            }

            if (result.IsError && message.MustSucceed)
            {
                throw new RedisException(string.Format("A critical operation failed {0}: {1}", message.Command, result.Error().Message));
            }
            return message;
        }
        internal virtual void ProcessCallbacks(object ctx, RedisResult result)
        {
            CompleteMessage((RedisMessage)ctx, result);
        }

        private RedisResult ReadSingleResult(ref RedisProtocolReader reader)
        {
            byte b = reader.ReadByteOrFail();
            switch ((char)b)
            {
                case '+':
                    return RedisResult.Message(reader.ReadBytesToCrlf());
                case '-':
                    return RedisResult.Error(reader.ReadStringToCrlf());
                case ':':
                    return RedisResult.Integer(reader.ReadInt64());
                case '$':
                    return RedisResult.Bytes(reader.ReadBulkBytes());
                case '*':
                    int count = (int)reader.ReadInt64();
                    if (count == -1) return RedisResult.Multi(null);
                    RedisResult[] inner = new RedisResult[count];
                    for (int i = 0; i < count; i++)
                    {
                        inner[i] = ReadSingleResult(ref reader);
                    }
                    return RedisResult.Multi(inner);
                default:
                    throw new RedisException("Not expecting header: &x" + b.ToString("x2"));
            }
        }
        internal void CompleteMessage(RedisMessage message, RedisResult result)
        {
            try
            {
                message.Complete(result);
            }
            catch (Exception ex)
            {
                OnError("Completing message", ex, false);
            }
        }
        private void Shutdown(string cause, Exception error)
        {
            LogToConsole("Shutdown: {0}, {1}", cause, error == null ? null : error.Message);
            Close(error != null);
            Interlocked.CompareExchange(ref state, (int)ConnectionState.Closed, (int)ConnectionState.Closing);

            if (error != null) OnError(cause, error, true);
            ShuttingDown(error);
            Dispose();
            var handler = Closed;
            if (handler != null) handler(this, EventArgs.Empty);

        }
        internal int GetSentCount() { lock (sent) { return sent.Count; } }

        /// <summary>
        /// Invoked when the server is terminating
        /// </summary>
        protected void ShuttingDown(Exception error)
        {
            RedisMessage message;
            RedisResult result = null;

            lock (sent)
            {
                if (sent.Count > 0)
                {
                    result = RedisResult.Error(
                        error == null ? "The server terminated before a reply was received"
                        : ("Error processing data: " + error.Message));
                }
                while (sent.Count > 0)
                { // notify clients of things that just didn't happen

                    message = sent.Dequeue();
                    CompleteMessage(message, result);
                }
            }
        }


 
        /// <summary>
        /// Indicates the number of commands executed on a per-database basis
        /// </summary>
        protected Dictionary<int, int> GetDbUsage()
        {
            lock (dbUsage)
            {
                return new Dictionary<int, int>(dbUsage);
            }
        }
        int messagesSent, messagesReceived, queueJumpers, messagesCancelled, errorMessages;
        private readonly Dictionary<int, int> dbUsage = new Dictionary<int, int>();
        private void LogUsage(int db)
        {
            lock (dbUsage)
            {
                int count;
                if (dbUsage.TryGetValue(db, out count))
                {
                    dbUsage[db] = count + 1;
                }
                else
                {
                    dbUsage.Add(db, 1);
                }
            }
        }
        /// <summary>
        /// Invoked when any error message is received on the connection.
        /// </summary>
        public event EventHandler<ErrorEventArgs> Error;
        /// <summary>
        /// Raises an error event
        /// </summary>
        protected void OnError(object sender, ErrorEventArgs args)
        {
            var handler = Error;
            if (handler != null)
            {
                handler(sender, args);
            }
        }
        /// <summary>
        /// Raises an error event
        /// </summary>
        protected void OnError(string cause, Exception ex, bool isFatal)
        {
            var handler = Error;
            var agg = ex as AggregateException;
            if (handler == null)
            {
                if (agg != null)
                {
                    foreach (var inner in agg.InnerExceptions)
                    {
                        Trace.WriteLine(inner.Message, cause);
                    }
                }
                else
                {
                    Trace.WriteLine(ex.Message, cause);
                }
            }
            else
            {
                if (agg != null)
                {
                    foreach (var inner in agg.InnerExceptions)
                    {
                        handler(this, new ErrorEventArgs(inner, cause, isFatal));
                    }
                }
                else
                {
                    handler(this, new ErrorEventArgs(ex, cause, isFatal));
                }
            }
        }

        internal void Flush()
        {
            asyncSocket.Flush();
            asyncSocket.PromptToWrite();
        }
        internal void WriteMessage(ref int db, RedisMessage next, IList<QueuedMessage> queued)
        {

            if (next.Db >= 0)
            {
                if (db != next.Db)
                {
                    db = next.Db;
                    RedisMessage changeDb = RedisMessage.Create(db, RedisLiteral.SELECT, db).ExpectOk().Critical();
                    if (queued != null)
                    {
                        queued.Add((QueuedMessage)(changeDb = new QueuedMessage(changeDb)));
                    }
                    RecordSent(changeDb);
                    changeDb.Write(asyncSocket, commandMap);
                    LogToConsole("Request: {0}", changeDb);
                    Interlocked.Increment(ref messagesSent);
                }
                LogUsage(db);
            }

            if (next.Command == RedisLiteral.SELECT)
            {
                // dealt with above; no need to send SELECT, SELECT
            }
            else
            {
                var mm = next as IMultiMessage;
                var tmp = next;
                if(queued != null)
                {
                    if(mm != null) throw new InvalidOperationException("Cannot perform composite operations (such as transactions) inside transactions");
                    queued.Add((QueuedMessage)(tmp = new QueuedMessage(tmp)));
                }

                if (mm == null)
                {
                    RecordSent(tmp);
                    tmp.Write(asyncSocket, commandMap);
                    LogToConsole("Request: {0}", tmp);
                    Interlocked.Increment(ref messagesSent);
                    switch (tmp.Command)
                    {
                        // scripts can change database
                        case RedisLiteral.EVAL:
                        case RedisLiteral.EVALSHA:
                        // transactions can be aborted without running the inner commands (SELECT) that have been written
                        case RedisLiteral.DISCARD:
                        case RedisLiteral.EXEC:
                            // we can't trust the current database; whack it
                            db = -1;
                            break;
                    }
                }
                else
                {
                    mm.Execute(this, ref db);
                }
            }
        }

        internal void WriteRaw(RedisMessage message)
        {
            if (message.Db >= 0) throw new ArgumentException("message", "WriteRaw cannot be used with db-centric messages");
            RecordSent(message);
            message.Write(asyncSocket, commandMap);
            LogToConsole("Request: {0}", message);
            Interlocked.Increment(ref messagesSent);
        }
        private readonly Queue<RedisMessage> sent = new Queue<RedisMessage>();
        internal void RecordSent(RedisMessage message, bool drainFirst = false)
        {
            lock (sent)
            {
                if (drainFirst && sent.Count != 0)
                {
                    // drain it down; the dequeuer will wake us
                    Monitor.Wait(sent);
                }
                sent.Enqueue(message);
            }
        }

        /// <summary>
        /// Indicates the current state of the connection to the server
        /// </summary>
        public enum ConnectionState
        {
            /// <summary>
            /// A connection that has not yet been innitialized
            /// </summary>
            Shiny,
            /// <summary>
            /// A connection that is in the process of opening
            /// </summary>
            Opening,
            /// <summary>
            /// An open connection
            /// </summary>
            Open,
            /// <summary>
            /// A connection that is in the process of closing
            /// </summary>
            Closing,
            /// <summary>
            /// A connection that is now closed and cannot be used
            /// </summary>
            Closed
        }

        internal Task<bool> ExecuteBoolean(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultBoolean();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task<long> ExecuteInt64(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultInt64();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task ExecuteVoid(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultVoid();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task<double> ExecuteDouble(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultDouble();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task<byte[]> ExecuteBytes(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultBytes();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task<RedisResult> ExecuteRaw(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultRaw();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task<string> ExecuteString(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultString();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }
        internal Task<long?> ExecuteNullableInt64(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultNullableInt64();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }
        internal Task<double?> ExecuteNullableDouble(RedisMessage message, bool queueJump, object state = null)
        {
            var msgResult = new MessageResultNullableDouble(state);
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }
        internal Task<byte[][]> ExecuteMultiBytes(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultMultiBytes();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task<string[]> ExecuteMultiString(RedisMessage message, bool queueJump, object state = null)
        {
            var msgResult = new MessageResultMultiString(state);
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal Task<KeyValuePair<byte[], double>[]> ExecutePairs(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultPairs();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }
        internal Task<Dictionary<string, byte[]>> ExecuteHashPairs(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultHashPairs();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }
        internal Task<Dictionary<string, string>> ExecuteStringPairs(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultStringPairs();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }
        internal Task<KeyValuePair<string, double>[]> ExecuteStringDoublePairs(RedisMessage message, bool queueJump)
        {
            var msgResult = new MessageResultStringDoublePairs();
            message.SetMessageResult(msgResult);
            EnqueueMessage(message, queueJump);
            return msgResult.Task;
        }

        internal void EnqueueMessage(RedisMessage message, bool queueJump)
        {
            unsent.Enqueue(message, queueJump);
            AsyncMessageSocket tmp = asyncSocket;
            if(tmp != null) tmp.PromptToWrite();
        }
        internal void CancelUnsent() {
            var all = unsent.DequeueAll();
            for (int i = 0; i < all.Length; i++)
            {
                RedisResult result = RedisResult.Cancelled;
                object ctx = ProcessReply(ref result, all[i]);
                ProcessCallbacks(ctx, result);
            }
        }
        internal RedisMessage[] DequeueAll() { return unsent.DequeueAll(); }
        /// <summary>
        /// If the task is not yet completed, blocks the caller until completion up to a maximum of SyncTimeout milliseconds.
        /// Once a task is completed, the result is returned.
        /// </summary>
        /// <param name="task">The task to wait on</param>
        /// <returns>The return value of the task.</returns>
        /// <exception cref="TimeoutException">If SyncTimeout milliseconds is exceeded.</exception>
        public T Wait<T>(Task<T> task)
        {
            Wait((Task)task, syncTimeout);
            return task.Result;
        }
        /// <summary>
        /// If the task is not yet completed, blocks the caller until completion up to a maximum of SyncTimeout milliseconds.
        /// Once a task is completed, the result is returned.
        /// </summary>
        /// <param name="task">The task to wait on</param>
        /// <param name="timeout">The duration (in milliseconds) to wait for the task to complete</param>
        /// <returns>The return value of the task.</returns>
        /// <exception cref="TimeoutException">If SyncTimeout milliseconds is exceeded.</exception>
        public T Wait<T>(Task<T> task, int timeout)
        {
            Wait((Task)task, timeout);
            return task.Result;
        }

        /// <summary>
        /// If true, then when using the Wait methods, information about the oldest outstanding message
        /// is included in the exception; this often points to a particular operation that was monopolising
        /// the connection
        /// </summary>
        public bool IncludeDetailInTimeouts { get; set; }

        /// <summary>
        /// If the task is not yet completed, blocks the caller until completion up to a maximum of SyncTimeout milliseconds.
        /// </summary>
        /// <param name="task">The task to wait on</param>
        /// <exception cref="TimeoutException">If SyncTimeout milliseconds is exceeded.</exception>
        /// <remarks>If an exception is throw, it is extracted from the AggregateException (unless multiple exceptions are found)</remarks>
        public void Wait(Task task)
        {
            Wait(task, syncTimeout);
        }
        /// <summary>
        /// If the task is not yet completed, blocks the caller until completion up to a maximum of SyncTimeout milliseconds.
        /// </summary>
        /// <param name="task">The task to wait on</param>
        /// <param name="timeout">The duration (in milliseconds) to wait for the task to complete</param>
        /// <exception cref="TimeoutException">If SyncTimeout milliseconds is exceeded.</exception>
        /// <remarks>If an exception is throw, it is extracted from the AggregateException (unless multiple exceptions are found)</remarks>
        public void Wait(Task task, int timeout)
        {
            if (task == null) throw new ArgumentNullException("task");
            try
            {
                if (!task.Wait(timeout))
                {
                    throw CreateTimeout();
                }
            }
            catch (AggregateException ex)
            {
                if (ex.InnerExceptions.Count == 1)
                {
                    throw ex.InnerExceptions[0];
                }
                throw;
            }
        }
#if LOG_TO_CONSOLE
        private static int nextConnIndex = 0;
        private readonly int connIndex = Interlocked.Increment(ref nextConnIndex);
#endif
        [Conditional("LOG_TO_CONSOLE")]
        internal void LogToConsole(string value)
        {
#if LOG_TO_CONSOLE
            Console.WriteLine("[" + connIndex + "," + Thread.CurrentThread.ManagedThreadId + "] " + value);
#else
            Console.WriteLine(value);
#endif
        }
        [Conditional("LOG_TO_CONSOLE")]
        internal void LogToConsole(string format, params object[] args)
        {
            LogToConsole(string.Format(format, args));
        }

        /// <summary>
        /// Give some information about the oldest incomplete (but sent) message on the server
        /// </summary>
        protected string GetTimeoutSummary()
        {
            RedisMessage msg;
            lock (sent)
            {
                if (sent.Count == 0) return null;
                msg = sent.Peek();
            }
            return msg.ToString();
        }

        private TimeoutException CreateTimeout()
        {
            if (State != ConnectionState.Open)
            {
                return new TimeoutException("The operation has timed out; the connection is not open");
            }
            if(IncludeDetailInTimeouts)
            {
                string compete = GetTimeoutSummary();
                if(!string.IsNullOrWhiteSpace(compete))
                {
                    string message = "The operation has timed out; possibly blocked by: " + compete;
                    return new TimeoutException(message);
                }
            }
            return new TimeoutException();
        }
        /// <summary>
        /// Waits for all of a set of tasks to complete, up to a maximum of SyncTimeout milliseconds.
        /// </summary>
        /// <param name="tasks">The tasks to wait on</param>
        /// <exception cref="TimeoutException">If SyncTimeout milliseconds is exceeded.</exception>
        public void WaitAll(params Task[] tasks)
        {
            if (tasks == null) throw new ArgumentNullException("tasks");
            if (!Task.WaitAll(tasks, syncTimeout))
            {
                throw CreateTimeout();
            }
        }
        /// <summary>
        /// Waits for any of a set of tasks to complete, up to a maximum of SyncTimeout milliseconds.
        /// </summary>
        /// <param name="tasks">The tasks to wait on</param>
        /// <returns>The index of a completed task</returns>
        /// <exception cref="TimeoutException">If SyncTimeout milliseconds is exceeded.</exception>        
        public int WaitAny(params Task[] tasks)
        {
            if (tasks == null) throw new ArgumentNullException("tasks");
            return Task.WaitAny(tasks, syncTimeout);
        }
        /// <summary>
        /// Add a continuation (a callback), to be executed once a task has completed
        /// </summary>
        /// <param name="task">The task to add a continuation to</param>
        /// <param name="action">The continuation to perform once completed</param>
        /// <returns>A new task representing the composed operation</returns>
        public Task ContinueWith<T>(Task<T> task, Action<Task<T>> action)
        {
            return task.ContinueWith(action, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
        }
        /// <summary>
        /// Add a continuation (a callback), to be executed once a task has completed
        /// </summary>
        /// <param name="task">The task to add a continuation to</param>
        /// <param name="action">The continuation to perform once completed</param>
        /// <returns>A new task representing the composed operation</returns>
        public Task ContinueWith(Task task, Action<Task> action)
        {
            return task.ContinueWith(action, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
        }

        /// <summary>
        /// What type of connection is this
        /// </summary>
        public ServerType ServerType { get; protected internal set; }


        internal bool RequestData(ref int db)
        {
            RedisMessage next;
            bool isHigh, shouldFlush;
            while (unsent.TryDequeue(out next, out isHigh, out shouldFlush))
            {
                if (abort)
                {
                    CompleteMessage(next, RedisResult.Error("The system aborted before this message was sent"));
                    continue;
                }
                if (!next.ChangeState(MessageState.NotSent, MessageState.Sent))
                {
                    // already cancelled; not our problem any more...
                    Interlocked.Increment(ref messagesCancelled);
                    continue;
                }
                if (isHigh) Interlocked.Increment(ref queueJumpers);
                WriteMessage(ref db, next, null);
                if (shouldFlush) Flush();
                return true; // only need to write one at a time
            }
            return false;
            //    Interlocked.CompareExchange(ref state, (int)ConnectionState.Closing, (int)ConnectionState.Open);
            //    if (asyncSocket != null)
            //    {
            //        var quit = RedisMessage.Create(-1, RedisLiteral.QUIT).ExpectOk().Critical();

            //        RecordSent(quit, !abort);
            //        quit.Write(asyncSocket);
            //        asyncSocket.Flush();
            //        Interlocked.Increment(ref messagesSent);
            //    }
            //}
            //catch (Exception ex)
            //{
            //    OnError("Outgoing queue", ex, true);
            //}
        }

        private readonly MemoryStream bodyBuffer = new MemoryStream();
        internal bool ProcessIncoming(byte[] buffer, int offset, int count)
        {
            try
            {
                if (count <= 0)
                {   // EOF
                    Shutdown("End of stream", null);
                    return false;
                }
                else
                {
                    var segment = new RedisProtocolReader(buffer, offset, count, asyncSocket, bodyBuffer);
                    bool isEof = true;
                    while (segment.HasLocalData)
                    {
                        RedisResult result = ReadSingleResult(ref segment);
                        LogToConsole("Response: {0}", result);
                        Interlocked.Increment(ref messagesReceived);
                        object ctx = ProcessReply(ref result);

                        if (result.IsError)
                        {
                            Interlocked.Increment(ref errorMessages);
                            OnError("Redis server", result.Error(), false);
                        }
                        try
                        {
                            ProcessCallbacks(ctx, result);
                        }
                        catch (Exception ex)
                        {
                            OnError("Processing callbacks", ex, false);
                        }
                        isEof = false;
                        
                        // segment is exhausted, but there is more buffered; refill the
                        // segment from the buffer
                        if (!segment.HasLocalData && segment.HasRemoteData)
                        {
                            isEof = !segment.ReadMoreSync();
                        }
                    }
                    if (isEof)
                    {   // EOF
                        Shutdown("End of stream", null);
                        return false;
                    }
                    else
                    {
                        return true;
                    }
                }
            }
            catch (Exception ex)
            {
                switch ((ConnectionState)state)
                {
                    case ConnectionState.Closed:
                    case ConnectionState.Closing:
                        // yeah, that is expected
                        break;
                    default:
                        Shutdown("Invalid inbound stream", ex);
                        break;
                }               
                return false;
            }
        }
    }

    /// <summary>
    /// What type of server does this represent
    /// </summary>
    public enum ServerType
    {
        /// <summary>
        /// The server is not yet connected, or is not recognised
        /// </summary>
        Unknown = 0,
        /// <summary>
        /// The server is a master node, suitable for read and write
        /// </summary>
        Master = 1,
        /// <summary>
        /// The server is a replication slave, suitable for read
        /// </summary>
        Slave = 2,
        /// <summary>
        /// The server is a sentinel, used for anutomated configuration
        /// and failover
        /// </summary>
        Sentinel = 3
    }
}

